package com.ruoyi.flink.source;/**
 * @program: ruoyi
 * @description: CustomRichParallelSourceFunction
 * @author: zengchen
 * @create: 2020-12-06 11:11
 **/

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

/**
 * @program: ruoyi
 * @description: CustomRichParallelSourceFunction
 * @author: zengchen
 * @create: 2020-12-06 11:11
 **/
public class CustomRichParallelSourceFunction extends RichParallelSourceFunction<Long> {

    private long count = 0L;
    private volatile boolean isRunning = true;

    @Override
    public void run(SourceContext ctx) throws Exception {
        while (isRunning && count < 1000) {
            synchronized (ctx.getCheckpointLock()) {
                ctx.collect(count);
                count++;
                Thread.sleep(1000);
            }
        }
    }

    @Override
    public void cancel() {
        isRunning = false;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
